package day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.sql.Timestamp;
import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 每个用户每5秒钟窗口的pv
 * @Author ZYX
 * @Date 2022/2/10 19:59
 * @Version 1.0
 */
public class Demo09 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new ClickSource())
                .keyBy(obj->obj.user)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new WindowResult())
                .print();

        env.execute();
    }

    public static class WindowResult extends ProcessWindowFunction<Event,String,String, TimeWindow> {
        @Override
        public void process(String value, Context context, Iterable<Event> elements, Collector<String> out) throws Exception {
            // 在窗口关闭的时候，触发调用
            // 迭代器参数中包含了窗口中所有的元素
            long start = context.window().getStart();
            long end = context.window().getEnd();
            // 迭代器里面共有多少元素
            long count = elements.spliterator().getExactSizeIfKnown();

            out.collect("用户 "+value+" 在窗口 "+new Timestamp(start)+" ~ "+new Timestamp(end) +" 中的PV次数"+count);
        }
    }

    public static class ClickSource implements SourceFunction<Event> {
        private boolean running = true;
        private String[] userArr = {"Alen","Bob","Carter","Dave","Eden","Frank","Gucci"};
        private String[] urlArr = {"./home","./cart","./fav","./user/1","./product/2"};
        private Random random = new Random();

        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            while (running){
                // collect 方法，向下游发送数据
                sourceContext.collect(
                        new Event(userArr[random.nextInt(userArr.length)],
                                urlArr[random.nextInt(urlArr.length)],
                                Calendar.getInstance().getTimeInMillis())
                );
                TimeUnit.SECONDS.sleep(1L);
            }
        }

        @Override
        public void cancel() {
            running = false;

        }
    }

    public static class Event{
        public String user;
        public String url;
        public Long timeStamp;

        public Event() {
        }

        public Event(String user, String url, Long timeStamp) {
            this.user = user;
            this.url = url;
            this.timeStamp = timeStamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "user='" + user + '\'' +
                    ", url='" + url + '\'' +
                    ", timeStamp=" + timeStamp +
                    '}';
        }
    }
    
}
